package org.apache.solr.ocean.core;

import java.io.IOException;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.apache.lucene.index.IndexCommit;
import org.apache.lucene.index.IndexDeletionPolicy;
import org.apache.lucene.index.IndexReader;
import org.apache.lucene.store.Directory;
import org.apache.solr.ocean.cache.Cache;
import org.apache.solr.ocean.cache.CacheKey;
import org.apache.solr.ocean.cache.DocSetCacheKey;
import org.apache.solr.ocean.core.Batch.SlaveBatch;
import org.apache.solr.ocean.util.ConcurrentTreeMap;
import org.apache.solr.ocean.util.Constants;
import org.apache.solr.ocean.util.IndexUtil;
import org.apache.solr.ocean.util.Timeout;
import org.apache.solr.ocean.util.Util;
import org.apache.solr.schema.IndexSchema;
import org.apache.solr.search.DocSet;

public abstract class DirectoryIndex extends Index {
	public static Logger log = Logger.getLogger(DirectoryIndex.class.getName());
	protected Cache cache;
	protected final ConcurrentTreeMap<Long, DirectoryIndexSnapshot> indexSnapshotMap = new ConcurrentTreeMap<Long, DirectoryIndexSnapshot>();
	//protected final ConcurrentTreeMap<Long, DirectoryIndexSnapshot> indexSnapshotMapByReaderVersion = new ConcurrentTreeMap<Long, DirectoryIndexSnapshot>();
	protected DirectoryIndexDeletionPolicy indexDeletionPolicy = new DirectoryIndexDeletionPolicy();
	protected IndexReader initialIndexReader;
	
	public DirectoryIndex(Long id, IndexSchema indexSchema, TransactionSystem.Category category) {
		super(id, indexSchema, category);
	}

	public DirectoryIndexSnapshot getLatestIndexSnapshot() {
		return indexSnapshotMap.lastValue();
	}

	public DirectoryIndexSnapshot getIndexSnapshot(Long snapshotId) {
		return indexSnapshotMap.get(snapshotId);
	}
  
	public IndexSnapshot initialize(Long snapshotId, List<SlaveBatch> deleteOnlySlaveBatches, TransactionSystem.Category category, IndexSchema indexSchema) throws Exception, IndexException, IOException {
		IndexReader indexReader = initialIndexReader;
		if (deleteOnlySlaveBatches == null || deleteOnlySlaveBatches.size() == 0) {
			createNewSnapshot(snapshotId, indexReader, getIndexSchema());
		} else {
			for (SlaveBatch slaveBatch : deleteOnlySlaveBatches) {
				if (slaveBatch.hasDeletes()) {
					applyDeletes(true, slaveBatch.getDeletes(), null, indexReader, indexSchema);
				}
				indexReader = indexReader.reopen();
				createNewSnapshot(slaveBatch.getId(), indexReader, slaveBatch.getIndexSchema());
			}
		}
		assert snapshotId.equals(indexSnapshotMap.lastKey());
		return indexSnapshotMap.lastValue();
	}
	
	protected void rollback() throws Exception {
		IndexCommit lastCommit = indexDeletionPolicy.getLastIndexCommitPoint();
		if (lastCommit != null) {
			lastCommit.delete();
		}
	}

	protected void onCommit() throws Exception {

	}
  
	private DirectoryIndexSnapshot getSnapshotByReaderVersion(long version) {
	  for (DirectoryIndexSnapshot indexSnapshot : indexSnapshotMap.values()) {
	    if (indexSnapshot.getIndexReaderVersion() == version) {
	      return indexSnapshot;
	    }
	  }
	  return null;
	}
	
	public class DirectoryIndexDeletionPolicy implements IndexDeletionPolicy {
		private IndexCommit lastCommit;

		public void onInit(List commits) throws IOException {
			onCommit(commits);
		}

		public IndexCommit getLastIndexCommitPoint() {
			return lastCommit;
		}

		public void onCommit(List commits) throws IOException {
			try {
				DirectoryIndex.this.onCommit();
				lastCommit = (IndexCommit) commits.get(commits.size() - 1);
				for (int x = 0; x < commits.size() - 1; x++) {
					IndexCommit indexCommitPoint = (IndexCommit) commits.get(x);
					//String segmentsFileName = indexCommitPoint.getSegmentsFileName();
					long readerVersion = indexCommitPoint.getVersion();
					DirectoryIndexSnapshot indexSnapshot = getSnapshotByReaderVersion(readerVersion);//indexSnapshotMapByReaderVersion.get(readerVersion);
					if (!indexSnapshot.hasRef()) {
						indexCommitPoint.delete();
						indexSnapshot.delete();
						indexSnapshotMap.remove(indexSnapshot.getSnapshotId());
						//indexSnapshotMapByReaderVersion.remove(indexSnapshot.getIndexReaderVersion());
					}
				}
			} catch (Exception exception) {
				throw Util.asIOException(exception);
			}
		}
	}

	public abstract class DirectoryIndexSnapshot extends IndexSnapshot {
		protected IndexReader indexReader;

		public DirectoryIndexSnapshot(Long snapshotId, IndexReader indexReader, IndexSchema schema) {
			super(snapshotId, schema);
			this.indexReader = indexReader;
		}

		public abstract Long getHighestDocumentId() throws IOException;

		public abstract Long getHighestSnapshotId() throws IOException;
		
		void delete() throws Exception {
			indexReader.close();
			indexSnapshotMap.remove(snapshotId);
			//indexSnapshotMapByReaderVersion.remove(getIndexReaderVersion());
		}
    
		public int maxDoc() {
      return indexReader.maxDoc();
    }
		
		public IndexReader getIndexReader() {
			return indexReader;
		}

		public long getIndexReaderVersion() {
			return indexReader.getVersion();
		}

		public boolean hasRef() throws Exception {
			return getCategory().getSnapshots().contains(snapshotId);
		}

		public DocSet getDocSet(DocSetCacheKey cacheKey, Timeout timeout) throws Exception {
			return cache.getDocSet(cacheKey, this, timeout);
		}

		public boolean containsKey(CacheKey cacheKey) {
			return cache.containsKey(cacheKey);
		}

		public Set<CacheKey> getCacheKeys() {
			return cache.getCacheKeys();
		}
	}

	public Set<CacheKey> getCacheKeys(List<? extends IndexSnapshot> indexSnapshots) {
		Set<CacheKey> allCacheKeys = new HashSet<CacheKey>();
		for (IndexSnapshot indexSnapshot : indexSnapshots) {
			allCacheKeys.addAll(indexSnapshot.getCacheKeys());
		}
		Set<DocSetCacheKey> docSetCacheKeys = new HashSet<DocSetCacheKey>(2000);
		//Set<FieldCacheKey> fieldCacheKeys = new HashSet<FieldCacheKey>(100);
		for (CacheKey cacheKey : allCacheKeys) {
			if (cacheKey instanceof DocSetCacheKey) {
				if (allContains(cacheKey, indexSnapshots)) {
					docSetCacheKeys.add((DocSetCacheKey) cacheKey);
				}
			} //else if (cacheKey instanceof FieldCacheKey) {
			//	fieldCacheKeys.add((FieldCacheKey) cacheKey);
			//}
		}
		Set<CacheKey> returnCacheKeys = new HashSet<CacheKey>();
		returnCacheKeys.addAll(docSetCacheKeys);
		//returnCacheKeys.addAll(fieldCacheKeys);
		return returnCacheKeys;
	}

	private static boolean allContains(CacheKey cacheKey, List<? extends IndexSnapshot> indexSnapshots) {
		for (IndexSnapshot indexSnapshot : indexSnapshots) {
			if (!indexSnapshot.containsKey(cacheKey)) {
				return false;
			}
		}
		return true;
	}
	/**
	public static void writeObject(String fileName, Serializable object, Directory directory) throws IOException {
		IndexOutput indexOutput = directory.createOutput(fileName);
		IndexOutputStream indexOutputStream = new IndexOutputStream(indexOutput);
		ObjectOutputStream objectOutput = new ObjectOutputStream(indexOutputStream);
		objectOutput.writeObject(object);
		objectOutput.close();
		indexOutput.close();
	}

	public static Serializable readObject(String fileName, Directory directory) throws IOException, ClassNotFoundException {
		IndexInput indexInput = directory.openInput(fileName);
		try {
			IndexInputStream inputStream = new IndexInputStream(indexInput);
			ObjectInputStream objectInputStream = new ObjectInputStream(inputStream);
			return (Serializable) objectInputStream.readObject();
		} finally {
			indexInput.close();
		}
	}
  **/
	public long getSize() throws IOException {
		Directory directory = getDirectory();
		return IndexUtil.getSize(directory);
	}

	public DeletesResult commitDeletes(Deletes deletes, Transaction transaction) throws IndexException, InterruptedException, IOException {
		IndexSnapshot indexSnapshot = getLatestIndexSnapshot();
		assert indexSnapshot != null;
		assert indexSnapshot.getSnapshotId().equals(transaction.getPreviousId());
		IndexReader previousIndexReader = indexSnapshot.getIndexReader();
		IndexReader newIndexReader = previousIndexReader.reopen();
		try {
			DeletesResult deletesResult = applyDeletes(true, deletes, null, newIndexReader, transaction.getIndexSchema());
			transaction.ready(this);
			if (transaction.go()) {
				if (deletesResult.getNumDeleted() > 0) {
					newIndexReader.flush();
				}
				Long snapshotID = transaction.getId();
				createNewSnapshot(snapshotID, newIndexReader, transaction.getIndexSchema());
				return deletesResult;
			} else {
				rollback();
				return null;
			}
		} catch (Throwable throwable) {
			log.log(Level.SEVERE, "", throwable);
			transaction.failed(this, throwable);
			return null;
		}
	}

	protected abstract DirectoryIndexSnapshot createNewSnapshot(Long snapshotId, IndexReader newIndexReader, IndexSchema schema) throws IOException;

	protected void registerSnapshot(DirectoryIndexSnapshot indexSnapshot) throws IOException {
		indexSnapshotMap.put(indexSnapshot.getSnapshotId(), indexSnapshot);
		//indexSnapshotMapByReaderVersion.put(indexSnapshot.getIndexReaderVersion(), indexSnapshot);
	}

	public abstract Directory getDirectory();
}
